package com.xs.client;

import java.util.Random;
import java.util.UUID;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;

import org.apache.log4j.Logger;

import com.alibaba.fastjson.JSON;
import com.xs.CommandType;
import com.xs.CustomProtcol;
import com.xs.SimpleDecoder;
import com.xs.SimpleEncoder;
import com.xs.client.entity.RpcFuture;
import com.xs.client.entity.RpcRequest;
import com.xs.client.entity.RpcResponse;
import com.xs.client.handler.ClientHandler;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.ChannelPromise;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;

public final class NettyClient {

	private static final Logger logger = Logger.getLogger(ClientHandler.class);

	private int port = 9999;

	private String host = "127.0.0.1";

	private EventLoopGroup workGroup = new NioEventLoopGroup();

	private CountDownLatch latch = new CountDownLatch(1);

	private ClientHandler handler = new ClientHandler();

	private CopyOnWriteArrayList<ClientHandler> connectedHandlers = new CopyOnWriteArrayList<>();

	/**
	 * 
	 */
	public NettyClient(String host, int port) {
		this.port = port;
		this.host = host;
		// start();
	}

	public void start() {

		new Thread(() -> {
			Bootstrap boot = new Bootstrap();
			boot.group(workGroup);
			boot.channel(NioSocketChannel.class);
			// boot.option(ChannelOption.TCP_NODELAY, true);
			boot.handler(new ChannelInitializer<SocketChannel>() {

				@Override
				protected void initChannel(SocketChannel ch) throws Exception {

					ChannelPipeline pipeline = ch.pipeline();
					// pipeline.addLast(new DelimiterBasedFrameDecoder(10,
					// Delimiters.lineDelimiter()));
					// pipeline.addLast(new StringEncoder());
					// pipeline.addLast(new StringDecoder());
					pipeline.addLast(new SimpleEncoder(RpcRequest.class));
					pipeline.addLast(new SimpleDecoder(RpcResponse.class));

					pipeline.addLast(handler);

				}
			});
			try {

				ChannelFuture f = boot.connect(host, port).sync();

				f.addListener(ff -> {
					if (ff.isSuccess()) {

						// this.handler =
						// f.channel().pipeline().get(ClientHandler.class);
						connectedHandlers.add(this.handler);
						latch.countDown();
					}
				});
				latch.await();

				logger.info("客户端：" + f.channel().localAddress() + " ， 启动成功");

				f.channel().closeFuture().sync();

				/*
				 * CustomProtcol msg = new CustomProtcol();
				 * while(!Thread.interrupted()){ String question = new
				 * Random().nextInt(50)+"加"+new Random().nextInt(50)+"=?";
				 * msg.setBody(question.getBytes()); msg.setCommand(1);
				 * Thread.sleep(1000); sendMsg(msg); }
				 */
				// f.channel().closeFuture().sync();

			} catch (InterruptedException e) {
				e.printStackTrace();
			} finally {
				workGroup.shutdownGracefully();
			}

		}).start();

	}

	public ClientHandler getHandler() {
		return handler;
	}

	public CopyOnWriteArrayList<ClientHandler> getConnectedHandlers() {
		return connectedHandlers;
	}

	public static void main(String[] args) throws InterruptedException, ExecutionException {
		for (int i = 0; i < 1; i++) {
			
			new Thread(() -> {
				

				
				NettyClient netty = new NettyClient("127.0.0.1", 8888);

				netty.start();

				CustomProtcol<RpcRequest> msg = new CustomProtcol<RpcRequest>();

				msg.setCommand(CommandType.QUESTION);

				while (true) {
					logger.debug("netty.connectedHandlers.size()=" + netty.connectedHandlers.size());
					if (netty.connectedHandlers.size() <= 0) {

						try {
							Thread.sleep(1000);
						} catch (InterruptedException e) {
							// TODO Auto-generated catch block
							e.printStackTrace();
						}

					} else {
						break;
					}
				}

				for (int n = 0; n < 5; n++) {

				//while(true){
					RpcRequest request = new RpcRequest();
					request.setRequestId(UUID.randomUUID().toString());
					String question = new Random().nextInt(50) + "加" + new Random().nextInt(50) + "=?";
					request.setBody(question);

					msg.setBody(request);

					RpcFuture obj;
					try {
						obj = netty.connectedHandlers.get(0).sendMessage(msg);
						logger.info("answer" + question + "=========RESULT=" + JSON.toJSONString(obj.getRpcResponse())
								+ "==================");
					} catch (InterruptedException | ExecutionException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}

					// logger.info("=========RpcFuture=" +
					// JSON.toJSONString(obj)+"==================");

				}

/*				RpcRequest request = new RpcRequest();
				request.setRequestId(UUID.randomUUID().toString());
				request.setBody(null);
				msg.setBody(request);
				msg.setCommand(CommandType.CLOSE);

				try {
					netty.connectedHandlers.get(0).sendMessage(msg);
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				} catch (ExecutionException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}*/

			}).start();
		}
	}

	public void benchTest(String[] args) throws InterruptedException, ExecutionException {

		NettyClient netty = new NettyClient("127.0.0.1", 8888);

		netty.start();

		CustomProtcol<RpcRequest> msg = new CustomProtcol<RpcRequest>();

		msg.setCommand(CommandType.QUESTION);

		while (true) {
			logger.info("netty.connectedHandlers.size()=" + netty.connectedHandlers.size());
			if (netty.connectedHandlers.size() <= 0) {

				Thread.sleep(1000);

			} else {
				break;
			}
		}

		for (int i = 0; i < 10; i++) {

			RpcRequest request = new RpcRequest();
			request.setRequestId(UUID.randomUUID().toString());
			String question = new Random().nextInt(50) + "加" + new Random().nextInt(50) + "=?";
			request.setBody(question);

			msg.setBody(request);

			RpcFuture obj = netty.connectedHandlers.get(0).sendMessage(msg);

			// logger.info("=========RpcFuture=" +
			// JSON.toJSONString(obj)+"==================");

			logger.info("answer" + question + "=========RESULT=" + JSON.toJSONString(obj.getRpcResponse())
					+ "==================");
		}

		/*
		 * RpcRequest request = new RpcRequest();
		 * request.setRequestId(UUID.randomUUID().toString());
		 * request.setBody(null); msg.setBody(request);
		 * msg.setCommand(CommandType.CLOSE);
		 * 
		 * netty.connectedHandlers.get(0).sendMessage(msg);
		 */
	}

	public Object send(CustomProtcol msg) throws InterruptedException, ExecutionException {
		RpcFuture rpcFuture = handler.sendMessage(msg);
		return rpcFuture.getRpcResponse();
	}

	/*
	 * public void sendMsg(CustomProtcol msg) { if (channel != null &&
	 * channel.isOpen()) {
	 * 
	 * ChannelFuture future = channel.writeAndFlush(msg); future.addListener(f
	 * -> {
	 * 
	 * if (f.isSuccess()) { logger.info(
	 * "===============================客户端发送消息成功==================================="
	 * ); } else { Throwable e = f.cause(); if (e != null) {
	 * e.printStackTrace(); } } });
	 * 
	 * } }
	 */
}
